package com.bjtu.boatmq.client.consumer;

import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Consumer {
	
	public static String ConsumerId;
	public static String GroupId;
	public static String RouterHost="127.0.0.1";
	public static int RouterPort=1234;
	public int HeartBeat;
	public static Map<String ,Set> Topic_Partition;
	public static List SubscribeList;
	public RouterConnector router_connector;
	public BrokerConnector broker_connector;
	public String router_data;
	public Consumer consumer;
	
	
	private ExecutorService executorService;
	
	Consumer(){
		this.ConsumerId="consumer001";
		this.GroupId="group01";
		this.HeartBeat=3000;

	}
	public void callback(String Msg) {
		System.out.println("从Router收到数据："+Msg);
	}
	
	public void Subscribe() {
		// TODO Auto-generated method stub
		
	}
	
	public void start() {
		
		//连接router
		router_connector=new RouterConnector(consumer, RouterHost,RouterPort,HeartBeat,ConsumerId);
		
		executorService = Executors.newFixedThreadPool(6);
		
		while (true) {
			router_data=router_connector.getMsg();
					if (router_data!= null)
						executorService.submit(new ConsumerThread(router_data));
		}
	}
	
	/**
	 * 
	 * @Title startFetch
	 * @Description TODO
	 * @author Dang
	 * @date 2020年6月9日 下午10:26:15  void
	 */
	public void startFetch() {
		//连接Broker
		//发起pull请求
		//开启线程，拉取消息
		//触发ConsumerListener
	}
	public void shutdown() {
		
		try {
			
			if (executorService != null) {
			executorService.shutdown();
			}
			if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
				System.out.println("Timeout");
				}
			} catch (InterruptedException ignored) {
				Thread.currentThread().interrupt();
				}
		}
	

}